package com.leon.raw_provider.tool;

import com.leon.rpc.JSONSerializer;
import com.leon.rpc.RpcDecoder;
import com.leon.rpc.RpcEncoder;
import com.leon.rpc.RpcRequest;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
public class RpcServer {

    @Value("${rpc.host}")
    private String host;

    @Value("${rpc.port}")
    private Integer port;

    private static String zkServer = "127.0.0.1:2181";

    @PostConstruct
    public void Init() {
        try {
            RpcServer.startServer(host, port);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void startServer(String host, Integer port) throws InterruptedException {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();

        bootstrap.group(eventLoopGroup).channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        /*pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new StringDecoder());*/
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new UserServiceHandler());
                    }
                });

        bootstrap.bind(host, port).sync();

        // 注册连接信息到 zookeeper 中
        registerZk(host, port);
    }

    // 注册连接信息到 zookeeper 中
    private static void registerZk(String host, Integer port) {
        String rootPath = "/raw_rpc";
        String serverNodePath = rootPath + "/server-node-";
        String serverInfo = host + ":" + port + ",0,0";    // 例:127.0.0.1:8989,0,0  第二个 0 为上次处理时间, 第三个 0 为上次响应时长,单位 ms
        ZkClient zkClient = new ZkClient(zkServer, 1000, 10000);

        boolean exists = zkClient.exists(rootPath);
        if (!exists) {
            zkClient.createPersistent(rootPath);
        }

        zkClient.createEphemeralSequential(serverNodePath, serverInfo);
    }


}
